package com.atguigu.gmall.realtime.util;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;

/**
 * @author Felix
 * @date 2022/8/9
 * 查询维度的工具类
 */
public class DimUtil {

    public static JSONObject getDimInfo(Connection conn, String tableName, String id){
        return getDimInfo(conn,tableName,Tuple2.of("ID",id));
    }

    /**
     * @param conn                  连接对象
     * @param tableName             要查询的维度表的表名
     * @param columnNameAndValues   查询条件
     * @return  查询出来的维度，将其封装为JSONObject
     * 查询维度，使用旁路缓存进行优化
     *  选型：
     *      flink的状态
     *          性能好，不便于维护
     *      Redis √
     *          性能也不差，方便维护
     *  思路：
     *      首先从缓存中查询维度数据，如何缓存中找到了要查询的维度数据，直接返回(缓存命中);
     *      如果缓存中没有找到要查询的维度数据，那么发送请求到phoenix表中查询维度，并将维度放到缓存中缓存起来
     *
     *  TTL: 1day
     *  Redis中存储数据类型：   String
     *  Redis中存储数据key:    dim:维度表名:主键1_主键2
     */
    public static JSONObject getDimInfo(Connection conn, String tableName, Tuple2<String,String> ... columnNameAndValues){
        //拼接从Redis中查询数据的key
        StringBuilder redisKey = new StringBuilder("dim:"+tableName.toLowerCase()+":");
        //拼接查询维度的SQL
        StringBuilder selectSql = new StringBuilder("select * from "+ GmallConfig.PHOENIX_SCHEMA +"." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue +"'");
            redisKey.append(columnValue);
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        //首先从缓存中查询维度数据
        Jedis jedis = null;
        String dimInfoStr = null;
        JSONObject dimJsonObj = null;

        try {
            jedis = RedisUtil.getJedis();
            dimInfoStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("~~~从Redis中查询维度数据发生了异常~~~");
        }

        if(StringUtils.isNotEmpty(dimInfoStr)){
            //如何缓存中找到了要查询的维度数据，直接返回(缓存命中)
            dimJsonObj = JSON.parseObject(dimInfoStr);
        }else{
            //如果缓存中没有找到要查询的维度数据，那么发送请求到phoenix表中查询维度
            System.out.println("从phoenix表中查询数据的SQL:" + selectSql);
            //查询维度，底层还是到phoenix表中去查询数据，所有调用的还是PhoenixUtil工具类中的查询方法
            List<JSONObject> dimJsonObjList = PhoenixUtil.queryList(conn, selectSql.toString(), JSONObject.class);
            if(dimJsonObjList != null && dimJsonObjList.size()>0){
                dimJsonObj = dimJsonObjList.get(0);
                //将查询出来的维度放到缓存中缓存起来
                if(jedis != null){
                    jedis.setex(redisKey.toString(),3600*24,dimJsonObj.toJSONString());
                }
            }else{
                System.out.println("~在phoenix表中没有找到对应的维度数据~");
            }
        }

        if(jedis != null){
            System.out.println("~~~关闭Jedis客户端~~~");
            jedis.close();
        }

        return dimJsonObj;
    }


    //根据维度的主键 到指定的维度表中查询维度数据出来  列=值
    public static JSONObject getDimInfoNoCache(Connection conn, String tableName, Tuple2<String,String> ... columnNameAndValues){
        //拼接查询维度的SQL
        StringBuilder selectSql = new StringBuilder("select * from "+ GmallConfig.PHOENIX_SCHEMA +"." + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue +"'");
            if(i < columnNameAndValues.length - 1){
                selectSql.append(" and ");
            }
        }

        System.out.println("从phoenix表中查询数据的SQL:" + selectSql);
        //查询维度，底层还是到phoenix表中去查询数据，所有调用的还是PhoenixUtil工具类中的查询方法
        List<JSONObject> dimJsonObjList = PhoenixUtil.queryList(conn, selectSql.toString(), JSONObject.class);
        JSONObject dimJsonObj = null;
        if(dimJsonObjList != null && dimJsonObjList.size()>0){
            dimJsonObj = dimJsonObjList.get(0);
        }else{
            System.out.println("~在phoenix表中没有找到对应的维度数据~");
        }
        return dimJsonObj;
    }

    //清除缓存中的维度数据
    public static void delCached(String tableName,String id){
        String redisKey = "dim:"+tableName.toLowerCase()+":" + id;
        Jedis jedis = null;
        try {
            jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(jedis != null){
                System.out.println("~~~清除缓存后，关闭Jedis客户端~~~");
                jedis.close();
            }
        }
    }

    public static void main(String[] args) throws SQLException {
        DruidDataSource dataSource = DruidDSUtil.createDataSource();
        DruidPooledConnection conn = dataSource.getConnection();

        // JSONObject dimJsonObj  = getDimInfoNoCache(conn, "DIM_BASE_TRADEMARK", Tuple2.of("id","1"));
        // JSONObject dimJsonObj  = getDimInfo(conn, "DIM_BASE_TRADEMARK", Tuple2.of("id","1"));
        JSONObject dimJsonObj  = getDimInfo(conn, "DIM_BASE_TRADEMARK", "1");
        System.out.println(dimJsonObj);
    }
}
